package ruc.DolphinDB.write;

import com.xxdb.DBConnection;
import com.xxdb.data.*;
import ruc.Consts.constant;
import ruc.DolphinDB.DolphinDB_Common;
import ruc.utils.IOUtils;

import java.io.IOException;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

public class DolphinDB_ConcurrentWrite {
    private int port;
    private String IP, tbName;
    private ReentrantLock lock = new ReentrantLock();

    public DolphinDB_ConcurrentWrite (String IP, int port, String tbName) {
        this.IP = IP;
        this.port = port;
        this.tbName = tbName;

    }

    private long splitAndSend(String[] arr, int all, int num) throws IOException {
        lock.lock();
        DBConnection conn = DolphinDB_Common.getDBConnection(IP, port);
        LinkedList<String> list = new LinkedList<>();
        List<Entity> args = null;
        List<Long> ts = new LinkedList<>();
        List<String>  userInfo = new LinkedList<>(), server = new LinkedList<>(), hostname = new LinkedList<>();
        List<Integer> cpu_usage = new LinkedList<>(), memory_usage = new LinkedList<>(), disk_usage = new LinkedList<>();
        try {
            for(int i = 0; i < arr.length; i++) {
                if (i % all != num) {
                    continue;
                }

                list.addLast(arr[i]);
            }

            // 批量插入
            for (String s : list) {
                String[] v = s.split(constant.Delimiter);

                ts.add(Long.parseLong(v[0]));
                userInfo.add(v[1]);
                server.add(v[2]);
                hostname.add(v[3]);
                cpu_usage.add(Integer.parseInt(v[4]));
                memory_usage.add(Integer.parseInt(v[5]));
                disk_usage.add(Integer.parseInt(v[6]));
            }
        } finally {
            lock.unlock();
        }

        args = Arrays.asList(new BasicLongVector(ts), //long 类型
                new BasicStringVector(userInfo), new BasicStringVector(server), new BasicStringVector(hostname), //string类型
                new BasicIntVector(cpu_usage), new BasicIntVector(memory_usage), new BasicIntVector(disk_usage)); //int类型

        long st_time = System.nanoTime();
        conn.run(String.format("tableInsert{%s}", tbName), args);
        return (System.nanoTime() - st_time) / constant.Ns2Ms;
    }

    private void MulInsertData(int[] ThreadNum, String[] val) throws InterruptedException {
        for(int i = 0; i < ThreadNum.length; i++) {
            CountDownLatch CDL = new CountDownLatch(ThreadNum[i]);
            AtomicLong costTime = new AtomicLong(0);
            for(int k = 0; k < ThreadNum[i]; k++) {
                int finalK = k, finalI = i;
                new Thread(()->{
                    try {
                        costTime.addAndGet(splitAndSend(val, ThreadNum[finalI], finalK));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    CDL.countDown();
                }).start();
            }

            CDL.await();
            System.out.printf("[TSBench-D] [DolphinDB_ConcurrentWrite] >>>>>> threadNum %d, insert time %d ms >>>>>> \n", ThreadNum[i], costTime.longValue());
            IOUtils.WriteData(constant.GetWriteFilePath("DolphinDB", "ConcurrentWrite"),
                                new String[]{String.valueOf(ThreadNum[i]), String.valueOf(costTime.longValue())});
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DolphinDB_ConcurrentWrite db = new DolphinDB_ConcurrentWrite(constant.IP, constant.DolphinDB_Port, constant.Concurrent_tbName);
        db.MulInsertData(constant.ConWrite_Thread, IOUtils.GetData());
    }
}
